






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 					_snake_case						(			_A ):
	_A          =   42




class 					_snake_case						(			_A       ,    _A ):
	_A          =   True






	@register_to_config
	def __init__(		self					,UpperCamelCase = 3					,UpperCamelCase = 3					,UpperCamelCase = ("DownEncoderBlock2D",)					,UpperCamelCase = ("UpDecoderBlock2D",)					,UpperCamelCase = (64,)					,UpperCamelCase = 1					,UpperCamelCase = "silu"					,UpperCamelCase = 4					,UpperCamelCase = 32					,UpperCamelCase = 32					,UpperCamelCase = 0.18215					,)						->    int:
					super().__init__()

					# pass init params to Encoder
					snake_case__      :Union[str, Any]         =   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
					snake_case__      :int         =   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					,)

					snake_case__      :Any         =   nn.Convad(2 * latent_channels					,2 * latent_channels					,1       )
					snake_case__      :List[Any]         =   nn.Convad(UpperCamelCase					,UpperCamelCase					,1       )

					snake_case__      :List[str]         =   False
					snake_case__      :List[str]         =   False

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






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase=False       )						->    str:
					if isinstance(UpperCamelCase					,(Encoder, Decoder)       ):
									snake_case__      :List[str]         =   value






	def    lowerCAmelCase_							(		self					,UpperCamelCase = True       )						->    List[str]:
					snake_case__      :Any         =   use_tiling






	def    lowerCAmelCase_							(		self       )						->    List[str]:
					self.enable_tiling(UpperCamelCase       )






	def    lowerCAmelCase_							(		self       )						->    Dict:
					snake_case__      :Any         =   True






	def    lowerCAmelCase_							(		self       )						->    Tuple:
					snake_case__      :Any         =   False






	@property
	# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
	def    lowerCAmelCase_							(		self       )						->    Dict[str, AttentionProcessor]:
					snake_case__      :List[Any]         =   {}

					def fn_recursive_add_processors(UpperCamelCase					,UpperCamelCase					,UpperCamelCase       ):
									if hasattr(UpperCamelCase					,"set_processor"       ):
													snake_case__      :int         =   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    lowerCAmelCase_							(		self					,UpperCamelCase       )						->    Optional[Any]:
					snake_case__      :List[str]         =   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(UpperCamelCase					,UpperCamelCase					,UpperCamelCase       ):
									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    lowerCAmelCase_							(		self       )						->    Optional[int]:
					self.set_attn_processor(AttnProcessor()       )






	@apply_forward_hook
	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = True       )						->    AutoencoderKLOutput:
					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:
									snake_case__      :Optional[int]         =   [self.encoder(UpperCamelCase       ) for x_slice in x.split(1       )]
									snake_case__      :Union[str, Any]         =   torch.cat(UpperCamelCase       )
					else:
									snake_case__      :Any         =   self.encoder(UpperCamelCase       )

					snake_case__      :str         =   self.quant_conv(UpperCamelCase       )
					snake_case__      :Dict         =   DiagonalGaussianDistribution(UpperCamelCase       )

					if not return_dict:
									return (posterior,)

					return AutoencoderKLOutput(latent_dist=UpperCamelCase       )






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = True       )						->    Union[DecoderOutput, torch.FloatTensor]:
					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       )

					snake_case__      :Dict         =   self.post_quant_conv(UpperCamelCase       )
					snake_case__      :List[str]         =   self.decoder(UpperCamelCase       )

					if not return_dict:
									return (dec,)

					return DecoderOutput(sample=UpperCamelCase       )






	@apply_forward_hook
	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = True       )						->    Union[DecoderOutput, torch.FloatTensor]:
					if self.use_slicing and z.shape[0] > 1:
									snake_case__      :str         =   [self._decode(UpperCamelCase       ).sample for z_slice in z.split(1       )]
									snake_case__      :Any         =   torch.cat(UpperCamelCase       )
					else:
									snake_case__      :List[Any]         =   self._decode(UpperCamelCase       ).sample

					if not return_dict:
									return (decoded,)

					return DecoderOutput(sample=UpperCamelCase       )






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase					,UpperCamelCase       )						->    Dict:
					snake_case__      :int         =   min(a.shape[2]					,b.shape[2]					,UpperCamelCase       )
					for y in range(UpperCamelCase       ):
									snake_case__      :str         =   a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
					return b






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase					,UpperCamelCase       )						->    Dict:
					snake_case__      :Union[str, Any]         =   min(a.shape[3]					,b.shape[3]					,UpperCamelCase       )
					for x in range(UpperCamelCase       ):
									snake_case__      :int         =   a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
					return b






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = True       )						->    AutoencoderKLOutput:
					snake_case__      :int         =   int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)       )
					snake_case__      :Optional[int]         =   int(self.tile_latent_min_size * self.tile_overlap_factor       )
					snake_case__      :Optional[int]         =   self.tile_latent_min_size - blend_extent

					# Split the image into 512x512 tiles and encode them separately.
					snake_case__      :List[Any]         =   []
					for i in range(0					,x.shape[2]					,UpperCamelCase       ):
									snake_case__      :List[str]         =   []
									for j in range(0					,x.shape[3]					,UpperCamelCase       ):
													snake_case__      :Union[str, Any]         =   x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
													snake_case__      :List[Any]         =   self.encoder(UpperCamelCase       )
													snake_case__      :Optional[Any]         =   self.quant_conv(UpperCamelCase       )
													row.append(UpperCamelCase       )
									rows.append(UpperCamelCase       )
					snake_case__      :Tuple         =   []
					for i, row in enumerate(UpperCamelCase       ):
									snake_case__      :Tuple         =   []
									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:
																	snake_case__      :List[Any]         =   self.blend_v(rows[i - 1][j]					,UpperCamelCase					,UpperCamelCase       )
													if j > 0:
																	snake_case__      :List[str]         =   self.blend_h(row[j - 1]					,UpperCamelCase					,UpperCamelCase       )
													result_row.append(tile[:, :, :row_limit, :row_limit]       )
									result_rows.append(torch.cat(UpperCamelCase					,dim=3       )       )

					snake_case__      :Tuple         =   torch.cat(UpperCamelCase					,dim=2       )
					snake_case__      :Dict         =   DiagonalGaussianDistribution(UpperCamelCase       )

					if not return_dict:
									return (posterior,)

					return AutoencoderKLOutput(latent_dist=UpperCamelCase       )






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = True       )						->    Union[DecoderOutput, torch.FloatTensor]:
					snake_case__      :Tuple         =   int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)       )
					snake_case__      :Any         =   int(self.tile_sample_min_size * self.tile_overlap_factor       )
					snake_case__      :List[str]         =   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.
					snake_case__      :List[Any]         =   []
					for i in range(0					,z.shape[2]					,UpperCamelCase       ):
									snake_case__      :Tuple         =   []
									for j in range(0					,z.shape[3]					,UpperCamelCase       ):
													snake_case__      :Optional[Any]         =   z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
													snake_case__      :Union[str, Any]         =   self.post_quant_conv(UpperCamelCase       )
													snake_case__      :Tuple         =   self.decoder(UpperCamelCase       )
													row.append(UpperCamelCase       )
									rows.append(UpperCamelCase       )
					snake_case__      :Optional[int]         =   []
					for i, row in enumerate(UpperCamelCase       ):
									snake_case__      :Optional[int]         =   []
									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:
																	snake_case__      :List[str]         =   self.blend_v(rows[i - 1][j]					,UpperCamelCase					,UpperCamelCase       )
													if j > 0:
																	snake_case__      :str         =   self.blend_h(row[j - 1]					,UpperCamelCase					,UpperCamelCase       )
													result_row.append(tile[:, :, :row_limit, :row_limit]       )
									result_rows.append(torch.cat(UpperCamelCase					,dim=3       )       )

					snake_case__      :Union[str, Any]         =   torch.cat(UpperCamelCase					,dim=2       )
					if not return_dict:
									return (dec,)

					return DecoderOutput(sample=UpperCamelCase       )






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = False					,UpperCamelCase = True					,UpperCamelCase = None					,)						->    Union[DecoderOutput, torch.FloatTensor]:
					snake_case__      :Any         =   sample
					snake_case__      :Optional[Any]         =   self.encode(UpperCamelCase       ).latent_dist
					if sample_posterior:
									snake_case__      :Dict         =   posterior.sample(generator=UpperCamelCase       )
					else:
									snake_case__      :int         =   posterior.mode()
					snake_case__      :Optional[int]         =   self.decode(UpperCamelCase       ).sample

					if not return_dict:
									return (dec,)

					return DecoderOutput(sample=UpperCamelCase       )







from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_sentencepiece_available


__UpperCAmelCase							:   Tuple							=  {}

try:
	if not is_sentencepiece_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__UpperCAmelCase							:   List[Any]							=  ["BartphoTokenizer"]

if TYPE_CHECKING:
	try:
		if not is_sentencepiece_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .tokenization_bartpho import BartphoTokenizer

else:
	import sys

	__UpperCAmelCase							:   Tuple							=  _LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)