






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 							SCREAMING_SNAKE_CASE__		(       UpperCamelCase_		):





					"""simple docstring"""






					a_			=	42



class 							SCREAMING_SNAKE_CASE__		(       UpperCamelCase_				,     UpperCamelCase_		):





					"""simple docstring"""






					a_			=	True



					@register_to_config
					def __init__(      self  :						int							,	__A  :						int = 3							,	__A  :						int = 3							,	__A  :						Tuple[str] = ("DownEncoderBlock2D",)							,	__A  :						Tuple[str] = ("UpDecoderBlock2D",)							,	__A  :						Tuple[int] = (6_4,)							,	__A  :						int = 1							,	__A  :						str = "silu"							,	__A  :						int = 4							,	__A  :						int = 3_2							,	__A  :						int = 3_2							,	__A  :						float = 0.1_8_2_1_5							,	):
						super().__init__()

						# pass init params to Encoder
						snake_case__				:   List[str]     =						Encoder(
						    in_channels=__A							,	out_channels=__A							,	down_block_types=__A							,	block_out_channels=__A							,	layers_per_block=__A							,	act_fn=__A							,	norm_num_groups=__A							,	double_z=__A							,	)

						# pass init params to Decoder
						snake_case__				:   Union[str, Any]     =						Decoder(
						    in_channels=__A							,	out_channels=__A							,	up_block_types=__A							,	block_out_channels=__A							,	layers_per_block=__A							,	norm_num_groups=__A							,	act_fn=__A							,	)

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

						snake_case__				:   str     =						False
						snake_case__				:   Dict     =						False

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



					def  _lowercase  (      self  :						int							,	__A  :						int							,	__A  :						int=False ):
						if isinstance(__A							,	(Encoder, Decoder) ):
							snake_case__				:   List[Any]     =						value



					def  _lowercase  (      self  :						Union[str, Any]							,	__A  :						bool = True ):
						snake_case__				:   int     =						use_tiling



					def  _lowercase  (      self  :						Tuple ):
						self.enable_tiling(__A )



					def  _lowercase  (      self  :						Optional[Any] ):
						snake_case__				:   str     =						True



					def  _lowercase  (      self  :						Tuple ):
						snake_case__				:   Optional[int]     =						False



					@property
					# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
					def  _lowercase  (      self  :						Optional[Any] ):
						snake_case__				:   Optional[Any]     =						{}

						def fn_recursive_add_processors(__A  :						str							,	__A  :						torch.nn.Module							,	__A  :						Dict[str, AttentionProcessor] ):
							if hasattr(__A							,	"set_processor" ):
								snake_case__				:   List[Any]     =						module.processor

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

							return processors

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

						return processors



					def  _lowercase  (      self  :						Union[str, Any]							,	__A  :						Union[AttentionProcessor, Dict[str, AttentionProcessor]] ):
						snake_case__				:   str     =						len(self.attn_processors.keys() )

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

						def fn_recursive_attn_processor(__A  :						str							,	__A  :						torch.nn.Module							,	__A  :						List[str] ):
							if hasattr(__A							,	"set_processor" ):
								if not isinstance(__A							,	__A ):
									module.set_processor(__A )
								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}'''							,	__A							,	__A )

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



					def  _lowercase  (      self  :						Any ):
						self.set_attn_processor(AttnProcessor() )



					@apply_forward_hook
					def  _lowercase  (      self  :						Optional[Any]							,	__A  :						torch.FloatTensor							,	__A  :						bool = True ):
						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(__A							,	return_dict=__A )

						if self.use_slicing and x.shape[0] > 1:
							snake_case__				:   int     =						[self.encoder(__A ) for x_slice in x.split(1 )]
							snake_case__				:   str     =						torch.cat(__A )
						else:
							snake_case__				:   Dict     =						self.encoder(__A )

						snake_case__				:   Optional[int]     =						self.quant_conv(__A )
						snake_case__				:   int     =						DiagonalGaussianDistribution(__A )

						if not return_dict:
							return (posterior,)

						return AutoencoderKLOutput(latent_dist=__A )



					def  _lowercase  (      self  :						Optional[Any]							,	__A  :						torch.FloatTensor							,	__A  :						bool = True ):
						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(__A							,	return_dict=__A )

						snake_case__				:   Optional[int]     =						self.post_quant_conv(__A )
						snake_case__				:   List[str]     =						self.decoder(__A )

						if not return_dict:
							return (dec,)

						return DecoderOutput(sample=__A )



					@apply_forward_hook
					def  _lowercase  (      self  :						str							,	__A  :						torch.FloatTensor							,	__A  :						bool = True ):
						if self.use_slicing and z.shape[0] > 1:
							snake_case__				:   int     =						[self._decode(__A ).sample for z_slice in z.split(1 )]
							snake_case__				:   Union[str, Any]     =						torch.cat(__A )
						else:
							snake_case__				:   Optional[Any]     =						self._decode(__A ).sample

						if not return_dict:
							return (decoded,)

						return DecoderOutput(sample=__A )



					def  _lowercase  (      self  :						List[Any]							,	__A  :						Union[str, Any]							,	__A  :						Dict							,	__A  :						Union[str, Any] ):
						snake_case__				:   List[str]     =						min(a.shape[2]							,	b.shape[2]							,	__A )
						for y in range(__A ):
							snake_case__				:   Optional[Any]     =						a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
						return b



					def  _lowercase  (      self  :						str							,	__A  :						int							,	__A  :						Union[str, Any]							,	__A  :						str ):
						snake_case__				:   Optional[Any]     =						min(a.shape[3]							,	b.shape[3]							,	__A )
						for x in range(__A ):
							snake_case__				:   Tuple     =						a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
						return b



					def  _lowercase  (      self  :						List[str]							,	__A  :						torch.FloatTensor							,	__A  :						bool = True ):
						snake_case__				:   List[Any]     =						int(self.tile_sample_min_size * (1 - self.tile_overlap_factor) )
						snake_case__				:   Union[str, Any]     =						int(self.tile_latent_min_size * self.tile_overlap_factor )
						snake_case__				:   Union[str, Any]     =						self.tile_latent_min_size - blend_extent

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

						snake_case__				:   Any     =						torch.cat(__A							,	dim=2 )
						snake_case__				:   Optional[Any]     =						DiagonalGaussianDistribution(__A )

						if not return_dict:
							return (posterior,)

						return AutoencoderKLOutput(latent_dist=__A )



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

						snake_case__				:   Any     =						torch.cat(__A							,	dim=2 )
						if not return_dict:
							return (dec,)

						return DecoderOutput(sample=__A )






					def  _lowercase  (      self  :						Union[str, Any]							,	__A  :						torch.FloatTensor							,	__A  :						bool = False							,	__A  :						bool = True							,	__A  :						Optional[torch.Generator] = None							,	):
						snake_case__				:   List[Any]     =						sample
						snake_case__				:   int     =						self.encode(__A ).latent_dist
						if sample_posterior:
							snake_case__				:   Union[str, Any]     =						posterior.sample(generator=__A )
						else:
							snake_case__				:   int     =						posterior.mode()
						snake_case__				:   int     =						self.decode(__A ).sample

						if not return_dict:
							return (dec,)

						return DecoderOutput(sample=__A )











from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


__lowerCamelCase      :    Tuple   		=   {
    """configuration_roberta_prelayernorm""": [
        """ROBERTA_PRELAYERNORM_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """RobertaPreLayerNormConfig""",
        """RobertaPreLayerNormOnnxConfig""",
    ],
}

try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__lowerCamelCase      :    Tuple   		=   [
	    """ROBERTA_PRELAYERNORM_PRETRAINED_MODEL_ARCHIVE_LIST""",
	    """RobertaPreLayerNormForCausalLM""",
	    """RobertaPreLayerNormForMaskedLM""",
	    """RobertaPreLayerNormForMultipleChoice""",
	    """RobertaPreLayerNormForQuestionAnswering""",
	    """RobertaPreLayerNormForSequenceClassification""",
	    """RobertaPreLayerNormForTokenClassification""",
	    """RobertaPreLayerNormModel""",
	    """RobertaPreLayerNormPreTrainedModel""",
	]

try:
	if not is_tf_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__lowerCamelCase      :    Union[str, Any]   		=   [
	    """TF_ROBERTA_PRELAYERNORM_PRETRAINED_MODEL_ARCHIVE_LIST""",
	    """TFRobertaPreLayerNormForCausalLM""",
	    """TFRobertaPreLayerNormForMaskedLM""",
	    """TFRobertaPreLayerNormForMultipleChoice""",
	    """TFRobertaPreLayerNormForQuestionAnswering""",
	    """TFRobertaPreLayerNormForSequenceClassification""",
	    """TFRobertaPreLayerNormForTokenClassification""",
	    """TFRobertaPreLayerNormMainLayer""",
	    """TFRobertaPreLayerNormModel""",
	    """TFRobertaPreLayerNormPreTrainedModel""",
	]

try:
	if not is_flax_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__lowerCamelCase      :    List[Any]   		=   [
	    """FlaxRobertaPreLayerNormForCausalLM""",
	    """FlaxRobertaPreLayerNormForMaskedLM""",
	    """FlaxRobertaPreLayerNormForMultipleChoice""",
	    """FlaxRobertaPreLayerNormForQuestionAnswering""",
	    """FlaxRobertaPreLayerNormForSequenceClassification""",
	    """FlaxRobertaPreLayerNormForTokenClassification""",
	    """FlaxRobertaPreLayerNormModel""",
	    """FlaxRobertaPreLayerNormPreTrainedModel""",
	]


if TYPE_CHECKING:
	from .configuration_roberta_prelayernorm import (
	    ROBERTA_PRELAYERNORM_PRETRAINED_CONFIG_ARCHIVE_MAP,
	    RobertaPreLayerNormConfig,
	    RobertaPreLayerNormOnnxConfig,
	)

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_roberta_prelayernorm import (
		    ROBERTA_PRELAYERNORM_PRETRAINED_MODEL_ARCHIVE_LIST,
		    RobertaPreLayerNormForCausalLM,
		    RobertaPreLayerNormForMaskedLM,
		    RobertaPreLayerNormForMultipleChoice,
		    RobertaPreLayerNormForQuestionAnswering,
		    RobertaPreLayerNormForSequenceClassification,
		    RobertaPreLayerNormForTokenClassification,
		    RobertaPreLayerNormModel,
		    RobertaPreLayerNormPreTrainedModel,
		)

	try:
		if not is_tf_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_tf_roberta_prelayernorm import (
		    TF_ROBERTA_PRELAYERNORM_PRETRAINED_MODEL_ARCHIVE_LIST,
		    TFRobertaPreLayerNormForCausalLM,
		    TFRobertaPreLayerNormForMaskedLM,
		    TFRobertaPreLayerNormForMultipleChoice,
		    TFRobertaPreLayerNormForQuestionAnswering,
		    TFRobertaPreLayerNormForSequenceClassification,
		    TFRobertaPreLayerNormForTokenClassification,
		    TFRobertaPreLayerNormMainLayer,
		    TFRobertaPreLayerNormModel,
		    TFRobertaPreLayerNormPreTrainedModel,
		)

	try:
		if not is_flax_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_flax_roberta_prelayernorm import (
		    FlaxRobertaPreLayerNormForCausalLM,
		    FlaxRobertaPreLayerNormForMaskedLM,
		    FlaxRobertaPreLayerNormForMultipleChoice,
		    FlaxRobertaPreLayerNormForQuestionAnswering,
		    FlaxRobertaPreLayerNormForSequenceClassification,
		    FlaxRobertaPreLayerNormForTokenClassification,
		    FlaxRobertaPreLayerNormModel,
		    FlaxRobertaPreLayerNormPreTrainedModel,
		)

else:
	import sys

	__lowerCamelCase      :    Dict   		=   _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)



