




from dataclasses import dataclass
from typing import Dict, Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder







@dataclass
class     __lowerCAmelCase (       _snake_case ):

			"""simple docstring"""







			snake_case_   		=       42







class     __lowerCAmelCase (       _snake_case			,				_snake_case ):

			"""simple docstring"""







			snake_case_   		=       True


			@register_to_config
			def __init__(      self			,						lowerCamelCase__ = 3			,						lowerCamelCase__ = 3			,						lowerCamelCase__ = ("DownEncoderBlock2D",)			,						lowerCamelCase__ = ("UpDecoderBlock2D",)			,						lowerCamelCase__ = (64,)			,						lowerCamelCase__ = 1			,						lowerCamelCase__ = "silu"			,						lowerCamelCase__ = 4			,						lowerCamelCase__ = 32			,						lowerCamelCase__ = 32			,						lowerCamelCase__ = 0.1_82_15			,						)    ->      List[Any]:



							'''simple docstring'''
							super().__init__()

							# pass init params to Encoder
							__lowerCamelCase		       =    Encoder(
							    in_channels=_UpperCamelCase			,						out_channels=_UpperCamelCase			,						down_block_types=_UpperCamelCase			,						block_out_channels=_UpperCamelCase			,						layers_per_block=_UpperCamelCase			,						act_fn=_UpperCamelCase			,						norm_num_groups=_UpperCamelCase			,						double_z=_UpperCamelCase			,						)

							# pass init params to Decoder
							__lowerCamelCase		       =    Decoder(
							    in_channels=_UpperCamelCase			,						out_channels=_UpperCamelCase			,						up_block_types=_UpperCamelCase			,						block_out_channels=_UpperCamelCase			,						layers_per_block=_UpperCamelCase			,						norm_num_groups=_UpperCamelCase			,						act_fn=_UpperCamelCase			,						)

							__lowerCamelCase		       =    nn.Convad(2 * latent_channels			,						2 * latent_channels			,						1							)
							__lowerCamelCase		       =    nn.Convad(_UpperCamelCase			,						_UpperCamelCase			,						1							)

							__lowerCamelCase		       =    False
							__lowerCamelCase		       =    False

							# only relevant if vae tiling is enabled
							__lowerCamelCase		       =    self.config.sample_size
							__lowerCamelCase		       =    (
							    self.config.sample_size[0]
							    if isinstance(self.config.sample_size			,						(list, tuple)							)
							    else self.config.sample_size
							)
							__lowerCamelCase		       =    int(sample_size / (2 ** (len(self.config.block_out_channels							) - 1))							)
							__lowerCamelCase		       =    0.25


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__=False							)    ->      List[str]:



							'''simple docstring'''
							if isinstance(_UpperCamelCase			,						(Encoder, Decoder)							):
											__lowerCamelCase		       =    value


			def       lowercase_    (      self			,						lowerCamelCase__ = True							)    ->      int:



							'''simple docstring'''
							__lowerCamelCase		       =    use_tiling


			def       lowercase_    (      self							)    ->      Dict:



							'''simple docstring'''
							self.enable_tiling(_UpperCamelCase							)


			def       lowercase_    (      self							)    ->      Optional[Any]:



							'''simple docstring'''
							__lowerCamelCase		       =    True


			def       lowercase_    (      self							)    ->      List[Any]:



							'''simple docstring'''
							__lowerCamelCase		       =    False


			@property
			# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
			def       lowercase_    (      self							)    ->      Dict[str, AttentionProcessor]:



							'''simple docstring'''
							__lowerCamelCase		       =    {}

							def fn_recursive_add_processors(lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__							):
											if hasattr(_UpperCamelCase			,						'set_processor'							):
															__lowerCamelCase		       =    module.processor

											for sub_name, child in module.named_children():
															fn_recursive_add_processors(f"""{name}.{sub_name}"""			,						_UpperCamelCase			,						_UpperCamelCase							)

											return processors

							for name, module in self.named_children():
											fn_recursive_add_processors(_UpperCamelCase			,						_UpperCamelCase			,						_UpperCamelCase							)

							return processors


			def       lowercase_    (      self			,						lowerCamelCase__							)    ->      List[Any]:



							'''simple docstring'''
							__lowerCamelCase		       =    len(self.attn_processors.keys()							)

							if isinstance(_UpperCamelCase			,						_UpperCamelCase							) and len(_UpperCamelCase							) != count:
											raise ValueError(
											    f"""A dict of processors was passed, but the number of processors {len(_UpperCamelCase							)} does not match the"""
											    f""" number of attention layers: {count}. Please make sure to pass {count} processor classes."""							)

							def fn_recursive_attn_processor(lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__							):
											if hasattr(_UpperCamelCase			,						'set_processor'							):
															if not isinstance(_UpperCamelCase			,						_UpperCamelCase							):
																			module.set_processor(_UpperCamelCase							)
															else:
																			module.set_processor(processor.pop(f"""{name}.processor"""							)							)

											for sub_name, child in module.named_children():
															fn_recursive_attn_processor(f"""{name}.{sub_name}"""			,						_UpperCamelCase			,						_UpperCamelCase							)

							for name, module in self.named_children():
											fn_recursive_attn_processor(_UpperCamelCase			,						_UpperCamelCase			,						_UpperCamelCase							)


			def       lowercase_    (      self							)    ->      Union[str, Any]:



							'''simple docstring'''
							self.set_attn_processor(AttnProcessor()							)


			@apply_forward_hook
			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = True							)    ->      AutoencoderKLOutput:



							'''simple docstring'''
							if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
											return self.tiled_encode(_UpperCamelCase			,						return_dict=_UpperCamelCase							)

							if self.use_slicing and x.shape[0] > 1:
											__lowerCamelCase		       =    [self.encoder(_UpperCamelCase							) for x_slice in x.split(1							)]
											__lowerCamelCase		       =    torch.cat(_UpperCamelCase							)
							else:
											__lowerCamelCase		       =    self.encoder(_UpperCamelCase							)

							__lowerCamelCase		       =    self.quant_conv(_UpperCamelCase							)
							__lowerCamelCase		       =    DiagonalGaussianDistribution(_UpperCamelCase							)

							if not return_dict:
											return (posterior,)

							return AutoencoderKLOutput(latent_dist=_UpperCamelCase							)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = True							)    ->      Union[DecoderOutput, torch.FloatTensor]:



							'''simple docstring'''
							if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
											return self.tiled_decode(_UpperCamelCase			,						return_dict=_UpperCamelCase							)

							__lowerCamelCase		       =    self.post_quant_conv(_UpperCamelCase							)
							__lowerCamelCase		       =    self.decoder(_UpperCamelCase							)

							if not return_dict:
											return (dec,)

							return DecoderOutput(sample=_UpperCamelCase							)


			@apply_forward_hook
			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = True							)    ->      Union[DecoderOutput, torch.FloatTensor]:



							'''simple docstring'''
							if self.use_slicing and z.shape[0] > 1:
											__lowerCamelCase		       =    [self._decode(_UpperCamelCase							).sample for z_slice in z.split(1							)]
											__lowerCamelCase		       =    torch.cat(_UpperCamelCase							)
							else:
											__lowerCamelCase		       =    self._decode(_UpperCamelCase							).sample

							if not return_dict:
											return (decoded,)

							return DecoderOutput(sample=_UpperCamelCase							)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__							)    ->      Any:



							'''simple docstring'''
							__lowerCamelCase		       =    min(a.shape[2]			,						b.shape[2]			,						_UpperCamelCase							)
							for y in range(_UpperCamelCase							):
											__lowerCamelCase		       =    a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
							return b


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__							)    ->      Dict:



							'''simple docstring'''
							__lowerCamelCase		       =    min(a.shape[3]			,						b.shape[3]			,						_UpperCamelCase							)
							for x in range(_UpperCamelCase							):
											__lowerCamelCase		       =    a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
							return b


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = True							)    ->      AutoencoderKLOutput:



							'''simple docstring'''
							__lowerCamelCase		       =    int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)							)
							__lowerCamelCase		       =    int(self.tile_latent_min_size * self.tile_overlap_factor							)
							__lowerCamelCase		       =    self.tile_latent_min_size - blend_extent

							# Split the image into 512x512 tiles and encode them separately.
							__lowerCamelCase		       =    []
							for i in range(0			,						x.shape[2]			,						_UpperCamelCase							):
											__lowerCamelCase		       =    []
											for j in range(0			,						x.shape[3]			,						_UpperCamelCase							):
															__lowerCamelCase		       =    x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
															__lowerCamelCase		       =    self.encoder(_UpperCamelCase							)
															__lowerCamelCase		       =    self.quant_conv(_UpperCamelCase							)
															row.append(_UpperCamelCase							)
											rows.append(_UpperCamelCase							)
							__lowerCamelCase		       =    []
							for i, row in enumerate(_UpperCamelCase							):
											__lowerCamelCase		       =    []
											for j, tile in enumerate(_UpperCamelCase							):
															# blend the above tile and the left tile
															# to the current tile and add the current tile to the result row
															if i > 0:
																			__lowerCamelCase		       =    self.blend_v(rows[i - 1][j]			,						_UpperCamelCase			,						_UpperCamelCase							)
															if j > 0:
																			__lowerCamelCase		       =    self.blend_h(row[j - 1]			,						_UpperCamelCase			,						_UpperCamelCase							)
															result_row.append(tile[:, :, :row_limit, :row_limit]							)
											result_rows.append(torch.cat(_UpperCamelCase			,						dim=3							)							)

							__lowerCamelCase		       =    torch.cat(_UpperCamelCase			,						dim=2							)
							__lowerCamelCase		       =    DiagonalGaussianDistribution(_UpperCamelCase							)

							if not return_dict:
											return (posterior,)

							return AutoencoderKLOutput(latent_dist=_UpperCamelCase							)


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = True							)    ->      Union[DecoderOutput, torch.FloatTensor]:



							'''simple docstring'''
							__lowerCamelCase		       =    int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)							)
							__lowerCamelCase		       =    int(self.tile_sample_min_size * self.tile_overlap_factor							)
							__lowerCamelCase		       =    self.tile_sample_min_size - blend_extent

							# Split z into overlapping 64x64 tiles and decode them separately.
							# The tiles have an overlap to avoid seams between tiles.
							__lowerCamelCase		       =    []
							for i in range(0			,						z.shape[2]			,						_UpperCamelCase							):
											__lowerCamelCase		       =    []
											for j in range(0			,						z.shape[3]			,						_UpperCamelCase							):
															__lowerCamelCase		       =    z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
															__lowerCamelCase		       =    self.post_quant_conv(_UpperCamelCase							)
															__lowerCamelCase		       =    self.decoder(_UpperCamelCase							)
															row.append(_UpperCamelCase							)
											rows.append(_UpperCamelCase							)
							__lowerCamelCase		       =    []
							for i, row in enumerate(_UpperCamelCase							):
											__lowerCamelCase		       =    []
											for j, tile in enumerate(_UpperCamelCase							):
															# blend the above tile and the left tile
															# to the current tile and add the current tile to the result row
															if i > 0:
																			__lowerCamelCase		       =    self.blend_v(rows[i - 1][j]			,						_UpperCamelCase			,						_UpperCamelCase							)
															if j > 0:
																			__lowerCamelCase		       =    self.blend_h(row[j - 1]			,						_UpperCamelCase			,						_UpperCamelCase							)
															result_row.append(tile[:, :, :row_limit, :row_limit]							)
											result_rows.append(torch.cat(_UpperCamelCase			,						dim=3							)							)

							__lowerCamelCase		       =    torch.cat(_UpperCamelCase			,						dim=2							)
							if not return_dict:
											return (dec,)

							return DecoderOutput(sample=_UpperCamelCase							)






			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = False			,						lowerCamelCase__ = True			,						lowerCamelCase__ = None			,						)    ->      Union[DecoderOutput, torch.FloatTensor]:



							'''simple docstring'''
							__lowerCamelCase		       =    sample
							__lowerCamelCase		       =    self.encode(_UpperCamelCase							).latent_dist
							if sample_posterior:
											__lowerCamelCase		       =    posterior.sample(generator=_UpperCamelCase							)
							else:
											__lowerCamelCase		       =    posterior.mode()
							__lowerCamelCase		       =    self.decode(_UpperCamelCase							).sample

							if not return_dict:
											return (dec,)

							return DecoderOutput(sample=_UpperCamelCase							)










import os
import shutil
from pathlib import Path
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ..utils import ONNX_EXTERNAL_WEIGHTS_NAME, ONNX_WEIGHTS_NAME, is_onnx_available, logging


if is_onnx_available():
		import onnxruntime as ort


__UpperCAmelCase								=						logging.get_logger(__name__)

__UpperCAmelCase								=						{
    'tensor(bool)': np.bool_,
    'tensor(int8)': np.inta,
    'tensor(uint8)': np.uinta,
    'tensor(int16)': np.intaa,
    'tensor(uint16)': np.uintaa,
    'tensor(int32)': np.intaa,
    'tensor(uint32)': np.uintaa,
    'tensor(int64)': np.intaa,
    'tensor(uint64)': np.uintaa,
    'tensor(float16)': np.floataa,
    'tensor(float)': np.floataa,
    'tensor(double)': np.floataa,
}



class   lowerCamelCase      :







		'''simple docstring'''



		def __init__(							self				,		_UpperCamelCase=None				,		**_UpperCamelCase   )	->							Dict:
							logger.info('`diffusers.OnnxRuntimeModel` is experimental and might change in the future.'   )
							UpperCAmelCase_    :      Any            =						model
							UpperCAmelCase_    :      int            =						kwargs.get('model_save_dir'				,		_UpperCamelCase   )
							UpperCAmelCase_    :      List[Any]            =						kwargs.get('latest_model_name'				,		_UpperCamelCase   )



		def __call__(							self				,		**_UpperCamelCase   )	->							str:
							UpperCAmelCase_    :      Optional[int]            =						{k: np.array(_UpperCamelCase   ) for k, v in kwargs.items()}
							return self.model.run(_UpperCamelCase				,		_UpperCamelCase   )



		@staticmethod
		def 					__UpperCAmelCase			(							_UpperCamelCase				,		_UpperCamelCase=None				,		_UpperCamelCase=None   )	->							List[Any]:
							if provider is None:
												logger.info('No onnxruntime provider specified, using CPUExecutionProvider'   )
												UpperCAmelCase_    :      List[str]            =						'CPUExecutionProvider'

							return ort.InferenceSession(_UpperCamelCase				,		providers=[provider]				,		sess_options=_UpperCamelCase   )



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		_UpperCamelCase = None				,		**_UpperCamelCase   )	->							Dict:
							UpperCAmelCase_    :      Any            =						file_name if file_name is not None else ONNX_WEIGHTS_NAME

							UpperCAmelCase_    :      Optional[Any]            =						self.model_save_dir.joinpath(self.latest_model_name   )
							UpperCAmelCase_    :      str            =						Path(_UpperCamelCase   ).joinpath(_UpperCamelCase   )
							try:
												shutil.copyfile(_UpperCamelCase				,		_UpperCamelCase   )
							except shutil.SameFileError:
												pass

							# copy external weights (for models >2GB)
							UpperCAmelCase_    :      Optional[Any]            =						self.model_save_dir.joinpath(_UpperCamelCase   )
							if src_path.exists():
												UpperCAmelCase_    :      List[Any]            =						Path(_UpperCamelCase   ).joinpath(_UpperCamelCase   )
												try:
																	shutil.copyfile(_UpperCamelCase				,		_UpperCamelCase   )
												except shutil.SameFileError:
																	pass



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		**_UpperCamelCase				,		)	->							List[str]:
							if os.path.isfile(_UpperCamelCase   ):
												logger.error(f"Provided path ({save_directory}) should be a directory, not a file"   )
												return

							os.makedirs(_UpperCamelCase				,		exist_ok=_UpperCamelCase   )

							# saving model weights/files
							self._save_pretrained(_UpperCamelCase				,		**_UpperCamelCase   )



		@classmethod
		def 					__UpperCAmelCase			(							cls				,		_UpperCamelCase				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		_UpperCamelCase = False				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		**_UpperCamelCase				,		)	->							List[str]:
							UpperCAmelCase_    :      List[str]            =						file_name if file_name is not None else ONNX_WEIGHTS_NAME
							# load model from local directory
							if os.path.isdir(_UpperCamelCase   ):
												UpperCAmelCase_    :      Union[str, Any]            =						OnnxRuntimeModel.load_model(
												    os.path.join(_UpperCamelCase				,		_UpperCamelCase   )				,		provider=_UpperCamelCase				,		sess_options=_UpperCamelCase   )
												UpperCAmelCase_    :      Tuple            =						Path(_UpperCamelCase   )
							# load model from hub
							else:
												# download model
												UpperCAmelCase_    :      List[str]            =						hf_hub_download(
												    repo_id=_UpperCamelCase				,		filename=_UpperCamelCase				,		use_auth_token=_UpperCamelCase				,		revision=_UpperCamelCase				,		cache_dir=_UpperCamelCase				,		force_download=_UpperCamelCase				,		)
												UpperCAmelCase_    :      Union[str, Any]            =						Path(_UpperCamelCase   ).parent
												UpperCAmelCase_    :      List[str]            =						Path(_UpperCamelCase   ).name
												UpperCAmelCase_    :      Union[str, Any]            =						OnnxRuntimeModel.load_model(_UpperCamelCase				,		provider=_UpperCamelCase				,		sess_options=_UpperCamelCase   )
							return cls(model=_UpperCamelCase				,		**_UpperCamelCase   )



		@classmethod
		def 					__UpperCAmelCase			(							cls				,		_UpperCamelCase				,		_UpperCamelCase = True				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		**_UpperCamelCase				,		)	->							Optional[int]:
							UpperCAmelCase_    :      List[str]            =						None
							if len(str(_UpperCamelCase   ).split('@'   )   ) == 2:
												UpperCAmelCase_       ,  UpperCAmelCase_    :      Tuple            =						model_id.split('@'   )

							return cls._from_pretrained(
							    model_id=_UpperCamelCase				,		revision=_UpperCamelCase				,		cache_dir=_UpperCamelCase				,		force_download=_UpperCamelCase				,		use_auth_token=_UpperCamelCase				,		**_UpperCamelCase				,		)






